package com.atguigu.edu.realtime220815.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime220815.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime220815.bean.ExamPaperBean;
import com.atguigu.edu.realtime220815.util.ClickHouseUtils;
import com.atguigu.edu.realtime220815.util.DateFormatUtil;
import com.atguigu.edu.realtime220815.util.KafkaUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @Author: rym
 * @Description: TODO
 * @DateTime: 2023/2/18 11:31
 **/
public class DwsExamPaperNumWindow {
    /*
    * jsonStr转换为jsonObj，将每个用户的总分、总时长、考试人数封装为javabean，分组，开窗，聚合
    * */
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        String topic = "dwd_exam_user_detail";
        String groupId = "DwsExamPaperNumWindow_group";
        FlinkKafkaConsumer<String> flinkKafkaConsumer = KafkaUtils.getFlinkKafkaConsumer(topic, groupId);
        DataStreamSource<String> kfkDs = env.addSource(flinkKafkaConsumer);

        SingleOutputStreamOperator<JSONObject> jsonOBjDS = kfkDs.map(JSON::parseObject);

        //{"id":"31595","paper_id":"168","user_id":"968","score":"16.7","duration_sec":"1084","date_id":"2023-02-18","ts":"1676700668"}
        SingleOutputStreamOperator<ExamPaperBean> beanDS = jsonOBjDS.map(
                new MapFunction<JSONObject, ExamPaperBean>() {
                    @Override
                    public ExamPaperBean map(JSONObject value) throws Exception {
                        String paper_id = value.getString("paper_id");
                        Double score = value.getDouble("score");
                        Long duration_sec = value.getLong("duration_sec");
                        String user_id = value.getString("user_id");
                        Long ts = value.getLong("ts");

                        ExamPaperBean bean = ExamPaperBean.builder()
                                .paperId(paper_id)
                                .score(score)
                                .duration_total(duration_sec)
                                .setUid(new HashSet<>(Collections.singleton(user_id)))
                                .ts(ts)
                                .build();
                        return bean;
                    }
                });

        SingleOutputStreamOperator<ExamPaperBean> withWaterDS = beanDS.assignTimestampsAndWatermarks(WatermarkStrategy.<ExamPaperBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<ExamPaperBean>() {
                    @Override
                    public long extractTimestamp(ExamPaperBean element, long recordTimestamp) {
                        return element.getTs()*1000;
                    }
                }));

        KeyedStream<ExamPaperBean, String> keyDS = withWaterDS.keyBy(value -> value.getPaperId());

        WindowedStream<ExamPaperBean, String, TimeWindow> windowDS = keyDS.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<ExamPaperBean> reduceDS = windowDS.reduce(
                new ReduceFunction<ExamPaperBean>() {
                    @Override
                    public ExamPaperBean reduce(ExamPaperBean value1, ExamPaperBean value2) throws Exception {
                        value1.setDuration_total(value1.getDuration_total() + value2.getDuration_total());
                        value1.setScore(value1.getScore() + value2.getScore());
                        value1.getSetUid().addAll(value2.getSetUid());
                        return value1;
                    }
                },
                new WindowFunction<ExamPaperBean, ExamPaperBean, String, TimeWindow>() {

                    @Override
                    public void apply(String s, TimeWindow window, Iterable<ExamPaperBean> input, Collector<ExamPaperBean> out) throws Exception {
                        long start = window.getStart();
                        long end = window.getEnd();
                        long now = System.currentTimeMillis();

                        for (ExamPaperBean examPaperBean : input) {
                            examPaperBean.setStt(DateFormatUtil.toYmdHms(start));
                            examPaperBean.setEdt(DateFormatUtil.toYmdHms(end));
                            examPaperBean.setTs(now);
                            examPaperBean.setCount_people((long) examPaperBean.getSetUid().size());
                            out.collect(examPaperBean);
                        }
                    }
                }
        );
        //reduceDS.print();

        SingleOutputStreamOperator<ExamPaperBean> withCourseIdDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<ExamPaperBean>("dim_test_paper") {
                    @Override
                    public void join(JSONObject dimInfoJsonObj, ExamPaperBean obj) {
                        String paper_title = dimInfoJsonObj.getString("paper_title".toUpperCase());
                        String course_id = dimInfoJsonObj.getString("course_id".toUpperCase());
                        obj.setPaperName(paper_title);
                        obj.setCourseId(course_id);
                    }

                    @Override
                    public String getKey(ExamPaperBean obj) {
                        return obj.getPaperId();
                    }
                }, 60, TimeUnit.SECONDS
        );
        //withCourseIdDS.print();
        SingleOutputStreamOperator<ExamPaperBean> withCourseNameDS = AsyncDataStream.unorderedWait(
                withCourseIdDS,
                new DimAsyncFunction<ExamPaperBean>("dim_course_info") {
                    @Override
                    public void join(JSONObject dimInfoJsonObj, ExamPaperBean obj) {
                        String course_name = dimInfoJsonObj.getString("course_name".toUpperCase());
                        obj.setCourseName(course_name);
                    }

                    @Override
                    public String getKey(ExamPaperBean obj) {
                        return obj.getCourseId();
                    }
                }, 60, TimeUnit.SECONDS
        );
        withCourseNameDS.print();

        withCourseNameDS.addSink(ClickHouseUtils.getClickHouseSinkFunction(
                "insert into dws_exam_paper_num_window values(?,?,?,?,?,?,?,?,?,?)"
        ));
        env.execute();
    }
}
